package com.wyg.epkafkastarter.consumer;

import com.wyg.epkafkastarter.config.ConsumerConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.BatchMessageListener;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.util.Assert;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: WangYongGan
 * @Date: 2022/5/18 11:51
 * @Description:基本消费者，通过start传入函数式接口 {@link BatchConsumer}运行实际的方法
 */
@Slf4j
public class BatchConsumerBase implements BatchMessageListener<String, String> {

    public ContainerProperties containerProps;//
    public BatchMessageListener<String, String> messageListener;//
    public BatchConsumer consumer;
    public Thread concurrentThread;
    int max;
    ThreadPoolExecutor threadPool;
    int n;


    @Override
    public void onMessage(List<ConsumerRecord<String, String>> records) {
        threadPool.execute(() -> {
            consumer.accept(records);
        });
    }

    public void init() {
        this.n = Runtime.getRuntime().availableProcessors() * 2;
        this.max = 1000;
        this.messageListener = this;
        this.threadPool = new ThreadPoolExecutor(n, n, 0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(max), new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public static void start(BatchConsumer consumer) {
    }

    ;

    /**
     * Description:
     *
     * @param messageListener 消息监听器
     * @param consumer        消费者方
     * @param containerProps  topic指定与其他配置
     * @return
     * @author WangYongGan
     **/
    public void start(BatchMessageListener<String, String> messageListener,
                      BatchConsumer consumer,
                      ContainerProperties containerProps) {
        this.consumer = consumer;
        this.containerProps = containerProps;
        this.messageListener = messageListener;
        init();
        Assert.notNull(messageListener, "消息监听器不能为Null");
        Assert.notNull(consumer, "消费者不能为Null");
        ///关键
        containerProps.setMessageListener(messageListener);
        KafkaMessageListenerContainer<String, String> container = ConsumerConfig.createContainer(containerProps);
        container.setBeanName("messageListenerContainer");
        //启动
        container.start();
    }
}
